/*
 * Copyright 2009-2010 by The Regents of the University of California
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * you may obtain a copy of the License from
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package edu.ucr.cs.dblab.acestore.operators;

import java.nio.ByteBuffer;

import edu.ucr.cs.dblab.acestore.api.IOperator;
import edu.ucr.cs.dblab.acestore.api.ISinglePrevOperatorFactory;
import edu.ucr.cs.dblab.acestore.api.PageAccessHelper;
import edu.ucr.cs.dblab.acestore.api.TupleInterpreter;
import edu.ucr.cs.dblab.acestore.base.ACEContext;
import edu.ucr.cs.dblab.acestore.base.LRUCacheAlgorithm;
import edu.ucr.cs.dblab.acestore.base.datatypes.ISerDeserializer;
import edu.ucr.cs.dblab.acestore.base.datatypes.IntegerSerDeserializer;
import edu.ucr.cs.dblab.acestore.base.datatypes.SchemaDescriptor;

public class DuplicateRemovalOperatorFactory implements
        ISinglePrevOperatorFactory {

    /*
     * (non-Javadoc)
     * 
     * @see
     * edu.ucr.cs.dblab.acestore.api.ISinglePrevOperatorFactory#createOperator
     * (edu.ucr.cs.dblab.acestore.base.ACEContext,
     * edu.ucr.cs.dblab.acestore.api.IOperator)
     */
    @Override
    public IOperator createOperator(final ACEContext ctx,
            final IOperator prevOperator) {
        return new IOperator() {

            // Page buffer for the input data from previous operator
            private ByteBuffer workingPage;
            private int workingRec;
            private int outPageCount;
            private int workingPageID;

            @Override
            public void open() throws Exception {
                prevOperator.open();
                workingRec = 0;
                outPageCount = 0;
                workingPageID = 0;
                workingPage = null;
            }

            @Override
            public void getNext(ByteBuffer page) throws Exception {
                // Only run once when starting the operator
                if (workingPage == null) {
                    workingPage = ByteBuffer.wrap(new byte[ctx.pageSize]);
                    prevOperator.getNext(workingPage);
                }

                int recInserted = 0;
                page.position(0);
                boolean loadNext = true;
                while (loadNext) {

                    int recCount = PageAccessHelper
                            .GetPageRecCount(workingPage);
                    for (; workingRec < recCount; workingRec++) {
                        if (recInserted > 0) {
                            if (equals(page, recInserted - 1, workingPage,
                                    workingRec)) {
                                int val = PageAccessHelper.GetRecValue(page,
                                        recInserted - 1);
                                val += PageAccessHelper.GetRecValue(
                                        workingPage, workingRec);
                                PageAccessHelper.SetRecValue(page, recInserted - 1,
                                        val);
                                continue;
                            }
                        }
                        // Insert into out page
                        int startOffset = PageAccessHelper.GetRecStartOffset(
                                workingPage, workingRec);
                        int len = PageAccessHelper.GetRecEndOffset(workingPage,
                                workingRec) - startOffset + 1;
                        int insertOffset = PageAccessHelper.GetRecEndOffset(
                                page, recInserted - 1) + 1;
                        if (insertOffset + len >= page.capacity() - 2 * INTSIZE
                                - RESERVED_END - recInserted * INTSIZE) {
                            // write out
                            loadNext = false;
                            break;
                        } else {
                            // insert
                            page.put(workingPage.array(), startOffset, len);
                            PageAccessHelper.SetRecEndOffset(page, recInserted,
                                    page.position() - 1);
                            recInserted++;          
                        }
                    }
                    
                    if (loadNext) {
                        // after processing an input data page
                        workingPageID = PageAccessHelper
                                .GetPageReference(workingPage);
                        if (workingPageID >= 0) {
                            prevOperator.getNext(workingPage);
                            workingRec = 0;
                        } else {
                            loadNext = false;
                        }
                    }
                }
                PageAccessHelper.SetPageRecCount(page, recInserted);
                if (workingPageID >= 0) {
                    outPageCount++;
                    PageAccessHelper.SetPageReference(page, outPageCount);
                } else {
                    PageAccessHelper.SetPageReference(page, -1);
                }
            }

            private boolean equals(ByteBuffer b1, int idx1, ByteBuffer b2,
                    int idx2) throws Exception {
                int startOffset1 = idx1 == 0 ? 0 : PageAccessHelper
                        .GetRecEndOffset(b1, idx1 - 1) + 1;
                int startOffset2 = idx2 == 0 ? 0 : PageAccessHelper
                        .GetRecEndOffset(b2, idx2 - 1) + 1;
                int fieldsCount1 = b1.getInt(startOffset1);
                int fieldsCount2 = b2.getInt(startOffset2);
                if (fieldsCount1 != fieldsCount2) {
                    throw new Exception(
                            "Failed to compare two records with different dimensionality.");
                }
                for (int i = 0; i < fieldsCount1 - 1; i++) {
                    int val1 = IntegerSerDeserializer.DESERIALIZE(b1.array(),
                            startOffset1 + INTSIZE + fieldsCount1 * INTSIZE + i
                                    * INTSIZE);
                    int val2 = IntegerSerDeserializer.DESERIALIZE(b2.array(),
                            startOffset2 + INTSIZE + fieldsCount2 * INTSIZE + i
                                    * INTSIZE);
                    if (val1 != val2)
                        return false;
                }
                return true;
            }
            
            private void printTuple(ByteBuffer buf, int idx){
                int startOffset = PageAccessHelper.GetRecStartOffset(buf, idx);
                int endOffset = PageAccessHelper.GetRecEndOffset(buf, idx);
                int len = endOffset - startOffset + 1;
                TupleInterpreter tupleInterpreter = new TupleInterpreter(buf.array(), startOffset, len);
                System.out.println("Processing: " + tupleInterpreter.toString());
            }

            @Override
            public void close() throws Exception {
                // TODO Auto-generated method stub

            }
        };
    }

    /**
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        ACEContext ctx = new ACEContext(256, 32, new LRUCacheAlgorithm());
        DataGeneratorOperatorFactory generatorFactory = new DataGeneratorOperatorFactory(
                20110602, 1, 200, (float) 0.5);
        IOperator generatorOperator = generatorFactory.createOperator(ctx);

        SortOperatorFactory sortFactory = new SortOperatorFactory(32,
                new SchemaDescriptor(new ISerDeserializer[] {
                        IntegerSerDeserializer.getSingleton(),
                        IntegerSerDeserializer.getSingleton() }), 0);
        IOperator sortOperator = sortFactory.createOperator(ctx,
                generatorOperator);

        DuplicateRemovalOperatorFactory dupFactory = new DuplicateRemovalOperatorFactory();
        IOperator dupOperator = dupFactory.createOperator(ctx, sortOperator);

        PagePrintOepratorFactory printerFactory = new PagePrintOepratorFactory();
        IOperator printerOperator = printerFactory.createOperator(ctx,
                dupOperator);

        ByteBuffer page = ByteBuffer.wrap(new byte[ctx.pageSize]);

        printerOperator.open();
        printerOperator.getNext(page);
        printerOperator.close();
    }
}
